/**
 * 
 */

/**
 * @author barise
 * @date   Dec 15, 2010
 * 
 */

package com.barisergun.memcached;

import static com.barisergun.memcached.TestVariables.*;

import java.io.IOException;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import net.spy.memcached.AddrUtil;
import net.spy.memcached.BinaryConnectionFactory;
import net.spy.memcached.MemcachedClient;

import org.testng.annotations.*;
import org.testng.*;

public class MemcachedTests {

	private MemcachedClient memcacheClient;

	private MemcachedProcess memcacheProcess;	

	
	@BeforeClass
	public void setup () throws IOException {

		memcacheProcess = new MemcachedProcess ();		

		List<List<String>> processesList = new LinkedList<List<String>> ();

		List<StringTokenizer> processesArguments = new LinkedList<StringTokenizer> ();

		for (int serverPortIncrement = 0; serverPortIncrement < NUMBER_OF_LOCAL_CACHE_SERVERS; serverPortIncrement++) {

			Integer portNumber = MEMCACHE_SERVER_ADDRESS_PORT_ROOT + serverPortIncrement;
			String portArgument = "-p" + portNumber.toString ();

			processesArguments.add (new StringTokenizer (String.format (
				PROCESS_STRING_FORMAT, MEMCACHE_SERVER_LOCAL_CACHE_SIZE_ARGUMENT, MEMCACHE_SERVER_LOCAL_ADDRESS, portArgument),
				DELIMETER_STRING));
		}

		for (StringTokenizer tokenizer : processesArguments) {

			List<String> arguments = new LinkedList<String> ();

			while (tokenizer.hasMoreTokens ()) {
				arguments.add (tokenizer.nextToken ());
			}

			processesList.add (arguments);

		}

		Assert.assertTrue (memcacheProcess.start (processesList));

		String addressesConcatenated = "";

		for (int serverPortIncrement = 0; serverPortIncrement < NUMBER_OF_LOCAL_CACHE_SERVERS; serverPortIncrement++) {

			String prefixSpaceDelimeter = serverPortIncrement > 0 ? " " : "";

			Integer portNumber = MEMCACHE_SERVER_ADDRESS_PORT_ROOT + serverPortIncrement;

			addressesConcatenated +=
				prefixSpaceDelimeter + MEMCACHE_SERVER_LOCAL_ADDRESS + ":" + portNumber.toString ();
		}

		memcacheClient =
			new MemcachedClient (new BinaryConnectionFactory (), AddrUtil
				.getAddresses (addressesConcatenated));
		processSystemProperties();

	}

	@AfterClass
	public void tearDown () throws InterruptedException {

		memcacheProcess.stop ();

		memcacheClient = null;

	}
	
	@Test
	public void testSimpleMemCache () throws InterruptedException {

		DataObject locData = new DataObject ();

		locData.setXValue (SAMPLE_X_COORD);
		locData.setZValue (SAMPLE_Y_COORD);

		Date dateNow = new Date ();
		locData.setTimeStamp (dateNow);

		memcacheClient.set (SINGLE_MSISDN_FOR_TEST, TTL_IN_SECONDS_SHORT, locData);

		DataObject objectFromCache =
			(DataObject) memcacheClient.get (SINGLE_MSISDN_FOR_TEST);

		Assert.assertEquals (objectFromCache.getTimeStamp (), locData.getTimeStamp ());
		Assert.assertEquals (objectFromCache.getXValue (), locData.getXValue ());
		Assert.assertEquals (objectFromCache.getZValue (), locData.getZValue ());

		Thread.sleep ((TTL_IN_SECONDS_SHORT - 1) * 1000);

		Object objectFromCacheAfterTimeout = memcacheClient.get (SINGLE_MSISDN_FOR_TEST);

		Assert.assertNotNull (objectFromCacheAfterTimeout);

		Thread.sleep ((TTL_IN_SECONDS_SHORT + 1) * 1000);

		objectFromCacheAfterTimeout = null;

		objectFromCacheAfterTimeout = memcacheClient.get (SINGLE_MSISDN_FOR_TEST);

		Assert.assertNull (objectFromCacheAfterTimeout);

	}

	@Test
	public void testWithManyCacheEntriesInSynchronizedStyle () throws InterruptedException,
			ExecutionException {

		Long startTime = System.currentTimeMillis ();

		DataObject[] locData = new DataObject[getEntryCount()];

		Map<String, Future<Boolean>> mapOfThreadFutures = new HashMap<String, Future<Boolean>> ();

		for (Integer fakeMsIsdn = 0; fakeMsIsdn < getEntryCount(); fakeMsIsdn++) {

			locData[fakeMsIsdn] = new DataObject ();

			Double xCoord = fakeMsIsdn / 10.00D;
			Double yCoord = fakeMsIsdn / 20.00D;

			locData[fakeMsIsdn].setXValue (xCoord);
			locData[fakeMsIsdn].setZValue (yCoord);
			Date dateNow = new Date ();
			locData[fakeMsIsdn].setTimeStamp (dateNow);

			mapOfThreadFutures.put (fakeMsIsdn.toString (), memcacheClient.set (fakeMsIsdn
				.toString (), getTtlInSeconds(), locData[fakeMsIsdn]));

			Assert.assertTrue (mapOfThreadFutures.get (fakeMsIsdn.toString ()).get (), String
				.format (FAILED_TO_INSERT_STRING, fakeMsIsdn));

		}

		Long endTime = System.currentTimeMillis ();

		Long executionTime = endTime - startTime;

		double tps = (getEntryCount() * 1000) / executionTime;

		;
		System.out.println (String.format (
			REPORT_INFO_STRING, MEMCACHED_SYNCH_TEST_INFO_STRING, INSERT_STRING, getEntryCount(),
			executionTime, NUMBER_OF_LOCAL_CACHE_SERVERS, tps));

		startTime = System.currentTimeMillis ();

		for (Integer fakeMsIsdn = 0; fakeMsIsdn < getEntryCount(); fakeMsIsdn++) {

			DataObject gotLocDataBack =
				(DataObject) memcacheClient.get (fakeMsIsdn.toString ());

			if (gotLocDataBack == null) {

				throw new NullPointerException (String.format (EXCEPTION_STRING, fakeMsIsdn
					.toString ()));
			}

			Assert.assertEquals (gotLocDataBack.getXValue (), locData[fakeMsIsdn].getXValue ());
			Assert.assertEquals (gotLocDataBack.getZValue (), locData[fakeMsIsdn].getZValue ());
			Assert.assertEquals (gotLocDataBack.getTimeStamp (), locData[fakeMsIsdn]
				.getTimeStamp ());

		}

		endTime = System.currentTimeMillis ();

		executionTime = endTime - startTime;

		tps = (getEntryCount() * 1000) / executionTime;

		System.out.println (String.format (
			REPORT_INFO_STRING, MEMCACHED_SYNCH_TEST_INFO_STRING, READING_STRING, getEntryCount(),
			executionTime, NUMBER_OF_LOCAL_CACHE_SERVERS, tps));

	}

	@Test
	public void testWithManyCacheEntriesInAsynchronizedStyle () throws InterruptedException,
			ExecutionException {

		Long startTime = System.currentTimeMillis ();

		DataObject[] locData = new DataObject[getEntryCount()];

		Map<String, Future<Boolean>> mapOfThreadFutures = new HashMap<String, Future<Boolean>> ();

		for (Integer fakeMsIsdn = 0; fakeMsIsdn < getEntryCount(); fakeMsIsdn++) {

			locData[fakeMsIsdn] = new DataObject ();

			Double xCoord = fakeMsIsdn / 10.00D;
			Double yCoord = fakeMsIsdn / 20.00D;
			locData[fakeMsIsdn].setXValue (xCoord);
			locData[fakeMsIsdn].setZValue (yCoord);
			Date dateNow = new Date ();
			locData[fakeMsIsdn].setTimeStamp (dateNow);
			mapOfThreadFutures.put (fakeMsIsdn.toString (), memcacheClient.set (fakeMsIsdn
				.toString (), getTtlInSeconds(), locData[fakeMsIsdn]));

		}

		Long endTime = System.currentTimeMillis ();

		Long executionTime = endTime - startTime;

		double tps = (getEntryCount() * 1000) / executionTime;

		System.out.println (String.format (
			REPORT_INFO_STRING, MEMCACHED_ASYNCH_TEST_INFO_STRING, INSERT_STRING, getEntryCount(),
			executionTime, NUMBER_OF_LOCAL_CACHE_SERVERS,tps));

		startTime = System.currentTimeMillis ();

		for (Entry<String, Future<Boolean>> futureEntry : mapOfThreadFutures.entrySet ()) {

			Assert.assertTrue (futureEntry.getValue ().get (), String.format (
				FAILED_TO_INSERT_STRING, futureEntry.getKey ()));

		}

		endTime = System.currentTimeMillis ();

		executionTime = endTime - startTime;

		System.out.println (String.format (
			REPORT_CONCURRENT_CACHES_TERMINATED_STRING, getEntryCount(), executionTime));

		startTime = System.currentTimeMillis ();

		for (Integer entryId = 0; entryId < getEntryCount(); entryId++) {

			DataObject gotLocDataBack =
				(DataObject) memcacheClient.get (entryId.toString ());

			if (gotLocDataBack == null) {

				throw new NullPointerException ("Exception in index " + entryId.toString ());
			}

			Assert.assertEquals (gotLocDataBack.getXValue (), locData[entryId].getXValue ());
			Assert.assertEquals (gotLocDataBack.getZValue (), locData[entryId].getZValue ());
			Assert.assertEquals (gotLocDataBack.getTimeStamp (), locData[entryId]
				.getTimeStamp ());

		}

		endTime = System.currentTimeMillis ();

		executionTime = endTime - startTime;

		tps = (getEntryCount() * 1000) / executionTime;

		System.out.println (String.format (
			REPORT_INFO_STRING, MEMCACHED_ASYNCH_TEST_INFO_STRING, READING_STRING, getEntryCount(),
			executionTime,NUMBER_OF_LOCAL_CACHE_SERVERS, tps));
	}
	
	@Test
	public void testPrependAppendOfMemcachedAndTTLMustBeReseted() throws InterruptedException, ExecutionException{
		
		String testString = "string";
		
		String prependString = "prepend ";
		
		String appendString = " append";
		
		memcacheClient.set (SINGLE_MSISDN_FOR_TEST, TTL_IN_SECONDS_SHORT, testString);

		String objectFromCache =
			(String) memcacheClient.get (SINGLE_MSISDN_FOR_TEST);

		Assert.assertEquals (objectFromCache , testString);
		
		Future<Boolean>  futureOfPrependOp = memcacheClient.prepend(0, SINGLE_MSISDN_FOR_TEST, prependString);
		
		Assert.assertTrue(futureOfPrependOp.get());
				
		String objectFromCacheAfterTimeout = (String) memcacheClient.get (SINGLE_MSISDN_FOR_TEST);
		
		Assert.assertEquals (objectFromCacheAfterTimeout , prependString + testString);
				
		Future<Boolean>  futureOfAppendOp = memcacheClient.append(0, SINGLE_MSISDN_FOR_TEST, appendString);
		
		Assert.assertTrue(futureOfAppendOp.get());
		
		Thread.sleep ((TTL_IN_SECONDS_SHORT - 1) * 1000);
		
		objectFromCacheAfterTimeout = (String) memcacheClient.get (SINGLE_MSISDN_FOR_TEST);
		
		Assert.assertEquals (objectFromCacheAfterTimeout , prependString + testString + appendString);

		Thread.sleep ((TTL_IN_SECONDS_SHORT + 1) * 1000);

		objectFromCacheAfterTimeout = null;

		objectFromCacheAfterTimeout = (String)memcacheClient.get (SINGLE_MSISDN_FOR_TEST);

		Assert.assertNull (objectFromCacheAfterTimeout);

		
		
	}
	
//	@Test
//	public void testPerformanceWithConcurrentHashmap () {
//
//		Long startTime = System.currentTimeMillis ();
//
//		Map<String, LocationDataObject> concurrentMap =
//			new ConcurrentHashMap<String, LocationDataObject> ();
//
//		LocationDataObject[] locData = new LocationDataObject[ENTRY_COUNT];
//
//		for (Integer fakeMsIsdn = 0; fakeMsIsdn < ENTRY_COUNT; fakeMsIsdn++) {
//
//			locData[fakeMsIsdn] = new LocationDataObject ();
//
//			Double xCoord = fakeMsIsdn / 10.00D;
//			Double yCoord = fakeMsIsdn / 20.00D;
//
//			locData[fakeMsIsdn].setXCoord (xCoord);
//			locData[fakeMsIsdn].setYCoord (yCoord);
//			Date dateNow = new Date ();
//			locData[fakeMsIsdn].setTimeStamp (dateNow);
//
//			concurrentMap.put (fakeMsIsdn.toString (), locData[fakeMsIsdn]);
//
//		}
//
//		Long endTime = System.currentTimeMillis ();
//
//		Long executionTime = endTime - startTime;
//
//		double tps = (ENTRY_COUNT * 1000) / executionTime;
//
//		System.out.println (String.format (
//			REPORT_INFO_STRING, CONCURRENT_HASHMAP_TEST_INFO_STRING, INSERT_STRING, ENTRY_COUNT,
//			executionTime, 1, tps));
//
//		startTime = System.currentTimeMillis ();
//
//		for (Integer fakeMsIsdn = 0; fakeMsIsdn < ENTRY_COUNT; fakeMsIsdn++) {
//
//			Assert.assertEquals (
//				concurrentMap.get (fakeMsIsdn.toString ()).getXCoord (), locData[fakeMsIsdn]
//					.getXCoord ());
//			Assert.assertEquals (
//				concurrentMap.get (fakeMsIsdn.toString ()).getYCoord (), locData[fakeMsIsdn]
//					.getYCoord ());
//			Assert.assertEquals (
//				concurrentMap.get (fakeMsIsdn.toString ()).getTimeStamp (), locData[fakeMsIsdn]
//					.getTimeStamp ());
//		}
//
//		endTime = System.currentTimeMillis ();
//
//		executionTime = endTime - startTime;
//
//		tps = (ENTRY_COUNT * 1000) / executionTime;
//
//		System.out.println (String.format (
//			REPORT_INFO_STRING, CONCURRENT_HASHMAP_TEST_INFO_STRING, READING_STRING, ENTRY_COUNT,
//			executionTime, 1, tps));
//
//	}

}
